package com.dx.chat.kafka;

import com.dx.chat.kafka.core.KafkaConfig;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.kafka.annotation.KafkaListenerConfigurationSelector;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.io.IOException;

/**
 * kafka自动加载，可针对性对某个消费组的消费者进行调优
 * 可以采用KafkaListener中groupid进行多消费者配置，ConcurrentKafkaListenerContainerFactory会默认共享
 *
 * @author wcx
 * @date 2023/6/28
 */
@RequiredArgsConstructor
@ComponentScan(basePackages = "com.dx.chat.kafka")
@EnableConfigurationProperties(value = KafkaConfig.class)
@ConditionalOnProperty(prefix = KafkaConfig.PREFIX, value = "enable", havingValue = "true", matchIfMissing = true)
@Import({KafkaListenerConfigurationSelector.class})
@Configuration(proxyBeanMethods = false)
public class KafkaMultiSourceAutoConfiguration {

    private final KafkaConfig kafkaConfig;

    @Bean
    @ConditionalOnProperty(name = "spring.kafka.producer.transaction-id-prefix")
    @ConditionalOnMissingBean
    public KafkaTransactionManager<?, ?> kafkaTransactionManager(ProducerFactory<?, ?> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }

    @Bean
    @ConditionalOnProperty(name = "spring.kafka.jaas.enabled")
    @ConditionalOnMissingBean
    public KafkaJaasLoginModuleInitializer kafkaJaasInitializer() throws IOException {
        KafkaJaasLoginModuleInitializer jaas = new KafkaJaasLoginModuleInitializer();
        KafkaConfig.Jaas jaasProperties = this.kafkaConfig.getJaas();
        if (jaasProperties.getControlFlag() != null) {
            jaas.setControlFlag(jaasProperties.getControlFlag());
        }
        if (jaasProperties.getLoginModule() != null) {
            jaas.setLoginModule(jaasProperties.getLoginModule());
        }
        jaas.setOptions(jaasProperties.getOptions());
        return jaas;
    }

}
